/**
 * libp2p 集群压力测试示例
 * 
 * 这个脚本演示 protoactor-ts 使用 libp2p 协议的集群在高负载下的表现
 * 它创建多个节点，每个节点上创建多个actor，并发送大量消息来测试系统性能
 */

import { Actor } from '../src/actor/actor';
import { PID } from '../src/actor/pid';
import { ActorSystem } from '../src/actor/actorSystem';
import { Props } from '../src/actor/props';
import { Context } from '../src/actor/context';
import { Remote, RemoteProtocolType } from '../src/remote/remote';
import { Cluster } from '../src/cluster/cluster';
import { ClusterConfig } from '../src/cluster/clusterConfig';
import { Libp2pProvider } from '../src/cluster/providers/libp2pProvider';
import { LocalIdentityLookup } from '../src/cluster/providers/localIdentityLookup';
import { ProcessRegistry } from '../src/actor/processRegistry';

// 配置参数
const CONFIG = {
  // 节点数量 - 增加到5个以提高分布式处理能力
  NODE_COUNT: 5,
  // 每个节点上创建的actor数量 - 增加到20个以提高并发处理能力
  ACTORS_PER_NODE: 20,
  // 每个actor发送的消息数量 - 增加到1000以获得更好的统计数据
  MESSAGES_PER_ACTOR: 1000,
  // 消息批次间隔(毫秒) - 降低到50ms以提高吞吐量
  BATCH_INTERVAL_MS: 50,
  // 消息发送超时(毫秒) - 保持合理的超时检测时间
  MESSAGE_TIMEOUT_MS: 3000,
  // 每批次发送的消息数量 - 增加到100以提高批处理效率
  BATCH_SIZE: 100,
  // 测试持续时间(毫秒) - 延长到5分钟以获得更稳定的性能数据
  TEST_DURATION_MS: 10,
  // 是否启用详细日志 - 关闭以减少日志开销
  VERBOSE_LOGGING: false
};

// 延时函数
const delay = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));

// 生成随机字符串
function randomString(length: number): string {
  const characters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
  let result = '';
  const charactersLength = characters.length;
  for (let i = 0; i < length; i++) {
    result += characters.charAt(Math.floor(Math.random() * charactersLength));
  }
  return result;
}

// ====== 定义工作Actor ======

// Actor消息类型
type WorkerActorMessage = 
  | { type: 'process', data: string, messageId: string }
  | { type: 'get-stats' };

// 工作Actor - 用于处理工作负载
class WorkerActor implements Actor {
  private processedCount: number = 0;
  private totalProcessingTime: number = 0;
  private errors: number = 0;
  private startTime: number = Date.now();
  
  constructor(private id: string) {
    if (CONFIG.VERBOSE_LOGGING) {
      console.log(`[WorkerActor:${this.id}] 创建`);
    }
  }
  
  async receive(context: Context): Promise<void> {
    try {
      const message = context.message();
      
      if (typeof message === 'object' && message !== null) {
        if (message.type === 'process') {
          const start = Date.now();
          
          if (CONFIG.VERBOSE_LOGGING) {
            console.log(`[WorkerActor:${this.id}] 处理消息: ${message.messageId}`);
          }
          
          // 模拟处理工作 - 随机延迟0-50ms
          const processingTime = Math.random() * 50;
          await delay(processingTime);
          
          this.processedCount++;
          const end = Date.now();
          this.totalProcessingTime += (end - start);
          
          // 创建响应
          const response = {
            success: true,
            messageId: message.messageId,
            processingTime,
            result: `处理完成: ${message.data}`
          };
          
          // 回复发送者
          const sender = context.sender();
          if (sender) {
            try {
              context.respond(response);
            } catch (error) {
              if (CONFIG.VERBOSE_LOGGING) {
                console.error(`[WorkerActor:${this.id}] 发送响应失败:`, error);
              }
              this.errors++;
            }
          }
        } else if (message.type === 'get-stats') {
          const uptime = Date.now() - this.startTime;
          
          // 计算统计信息
          const stats = {
            id: this.id,
            processedCount: this.processedCount,
            errors: this.errors,
            uptime,
            avgProcessingTime: this.processedCount > 0 ? this.totalProcessingTime / this.processedCount : 0,
            throughput: uptime > 0 ? (this.processedCount / uptime) * 1000 : 0 // 每秒处理的消息数
          };
          
          context.respond(stats);
        }
      }
    } catch (error) {
      console.error(`[WorkerActor:${this.id}] 处理消息时出错:`, error);
      this.errors++;
      
      const errorResponse = {
        success: false,
        error: error instanceof Error ? error.message : String(error)
      };
      
      const sender = context.sender();
      if (sender) {
        try {
          context.respond(errorResponse);
        } catch (respondError) {
          if (CONFIG.VERBOSE_LOGGING) {
            console.error(`[WorkerActor:${this.id}] 发送错误响应失败:`, respondError);
          }
        }
      }
    }
  }
}

// ====== 定义监控Actor ======

// 监控Actor - 用于收集性能指标
class MonitorActor implements Actor {
  private nodeStats: Map<string, any> = new Map();
  private startTime: number = Date.now();
  private messagesSent: number = 0;
  private messagesReceived: number = 0;
  private errors: number = 0;
  private latencies: number[] = [];
  
  constructor() {
    console.log(`[MonitorActor] 创建`);
  }
  
  async receive(context: Context): Promise<void> {
    const message = context.message();
    
    if (typeof message === 'object' && message !== null) {
      if (message.type === 'message-sent') {
        this.messagesSent++;
      } else if (message.type === 'message-received') {
        this.messagesReceived++;
        // 记录延迟
        if (message.latency) {
          this.latencies.push(message.latency);
        }
      } else if (message.type === 'error') {
        this.errors++;
      } else if (message.type === 'node-stats') {
        this.nodeStats.set(message.nodeId, message.stats);
      } else if (message.type === 'print-stats') {
        const stats = this.calculateStats();
        console.log('\n========== 集群性能统计 ==========');
        console.log(`测试运行时间: ${stats.uptimeSeconds.toFixed(2)}秒`);
        console.log(`消息发送数量: ${stats.messagesSent}`);
        console.log(`消息接收数量: ${stats.messagesReceived}`);
        console.log(`错误数量: ${stats.errors}`);
        console.log(`平均延迟: ${stats.avgLatency.toFixed(2)}ms`);
        console.log(`最小延迟: ${stats.minLatency.toFixed(2)}ms`);
        console.log(`最大延迟: ${stats.maxLatency.toFixed(2)}ms`);
        console.log(`吞吐量: ${stats.throughput.toFixed(2)} 消息/秒`);
        console.log('==================================\n');
        
        // 回复统计信息
        context.respond(stats);
      }
    }
  }
  
  private calculateStats() {
    const uptime = Date.now() - this.startTime;
    const uptimeSeconds = uptime / 1000;
    
    // 计算延迟统计
    let minLatency = this.latencies.length > 0 ? Math.min(...this.latencies) : 0;
    let maxLatency = this.latencies.length > 0 ? Math.max(...this.latencies) : 0;
    let avgLatency = this.latencies.length > 0 
      ? this.latencies.reduce((sum, latency) => sum + latency, 0) / this.latencies.length 
      : 0;
    
    // 计算吞吐量
    const throughput = uptimeSeconds > 0 ? this.messagesReceived / uptimeSeconds : 0;
    
    return {
      uptime,
      uptimeSeconds,
      messagesSent: this.messagesSent,
      messagesReceived: this.messagesReceived,
      errors: this.errors,
      minLatency,
      maxLatency,
      avgLatency,
      throughput
    };
  }
}

// ====== 节点类 ======

class ClusterNode {
  private name: string;
  private port: number;
  private system!: ActorSystem;
  private remote!: Remote;
  private libp2pProvider!: Libp2pProvider;
  private workerActors: PID[] = [];
  private monitorPid: PID | null = null;
  private running: boolean = false;
  private messagesSent: number = 0;
  private messagesReceived: number = 0;
  private errors: number = 0;
  
  constructor(name: string, port: number, private bootstrapAddrs: string[] = [], private isCoordinator: boolean = false) {
    this.name = name;
    this.port = port;
  }
  
  async start(): Promise<void> {
    if (this.running) {
      console.log(`${this.name} 已经在运行中...`);
      return;
    }
    
    console.log(`启动 ${this.name}...`);
    
    // 创建 Actor 系统
    this.system = new ActorSystem();
    
    // 订阅系统事件
    this.system.eventStream.subscribe('DeadLetter', (msg: any) => {
      console.log(`[${this.name}] DeadLetter: ${JSON.stringify(msg)}`);
    });
    
    // 初始化 libp2p Provider
    this.libp2pProvider = new Libp2pProvider({
      listenAddrs: [`/ip4/0.0.0.0/tcp/${this.port}`],
      bootstrapAddrs: this.bootstrapAddrs,
      enableMDNS: true,
      enableDHT: true,
      enableGossip: true
    });
    
    // 初始化 Remote
    this.remote = new Remote(this.system);
    this.remote
      .withProtocol(RemoteProtocolType.Libp2p)
      .withLibp2pProvider(this.libp2pProvider);
    
    // 创建集群配置
    const cluster = new Cluster(this.system, this.remote);
    
    // 初始化并启动 libp2p 提供者
    await this.libp2pProvider.init(cluster);
    await this.libp2pProvider.start();
    
    // 获取 peerId 并设置为本地地址
    const peerId = this.libp2pProvider.getPeerId();
    if (!peerId) {
      throw new Error(`${this.name} 无法获取 peerId`);
    }
    
    const localAddress = `libp2p:${peerId}`;
    console.log(`${this.name} 地址: ${localAddress}`);
    ProcessRegistry.getInstance().address = () => localAddress;
    
    // 注册Actor类型
    this.remote.register('worker', Props.fromProducer(() => new WorkerActor(`${this.name}-worker-default`)));
    
    if (this.isCoordinator) {
      // 创建监控actor
      const monitorProps = Props.fromProducer(() => new MonitorActor());
      this.monitorPid = this.system.root.spawn(monitorProps);
      console.log(`${this.name} 监控Actor已创建: ${this.monitorPid.toString()}`);
    }
    
    // 启动Remote
    const [_, hostname] = localAddress.split(':');
    await this.remote.start(hostname, this.port);
    console.log(`${this.name} Remote已启动，地址: ${localAddress}`);
    
    // 创建工作Actor
    for (let i = 0; i < CONFIG.ACTORS_PER_NODE; i++) {
      const actorId = `${i + 1}`;
      const workerProps = Props.fromProducer(() => new WorkerActor(`${this.name}-worker-${actorId}`));
      const pid = this.system.root.spawn(workerProps);
      this.workerActors.push(pid);
      if (CONFIG.VERBOSE_LOGGING) {
        console.log(`${this.name} 创建工作Actor ${actorId}: ${pid.toString()}`);
      }
    }
    
    this.running = true;
    console.log(`${this.name} 启动完成，共有 ${this.workerActors.length} 个工作Actor`);
  }
  
  async shutdown(): Promise<void> {
    console.log(`关闭节点: ${this.name}`);
    this.running = false;
    
    // 停止Remote
    if (this.remote) {
      await this.remote.shutdown();
    }
    
    // 停止libp2p
    if (this.libp2pProvider) {
      await this.libp2pProvider.shutdown();
    }
    
    console.log(`节点 ${this.name} 已关闭`);
  }
  
  getActors(): PID[] {
    return this.workerActors;
  }
  
  getAddress(): string {
    return ProcessRegistry.getInstance().address();
  }
  
  getMonitorPid(): PID | null {
    return this.monitorPid;
  }
  
  getSystem(): ActorSystem {
    return this.system;
  }
  
  isRunning(): boolean {
    return this.running;
  }
  
  // 向指定actor发送工作消息
  async sendWorkMessage(targetPid: PID, dataSize: number = 100): Promise<any> {
    if (!this.running) return null;
    
    const messageId = `msg-${Date.now()}-${randomString(8)}`;
    const data = randomString(dataSize);
    const message = {
      type: 'process',
      data,
      messageId
    };
    
    const sendTime = Date.now();
    
    try {
      if (CONFIG.VERBOSE_LOGGING) {
        console.log(`${this.name} 发送消息: ${messageId} 到 ${targetPid.toString()}`);
      }
      
      // 通知监控Actor消息已发送
      if (this.monitorPid) {
        this.system.root.send(this.monitorPid, { type: 'message-sent', messageId });
      }
      
      this.messagesSent++;
      const response = await this.system.root.requestFuture(
        targetPid, 
        message,
        CONFIG.MESSAGE_TIMEOUT_MS
      ).toPromise();
      
      const receiveTime = Date.now();
      const latency = receiveTime - sendTime;
      
      if (CONFIG.VERBOSE_LOGGING) {
        console.log(`${this.name} 收到响应: ${messageId}, 延迟: ${latency}ms`);
      }
      
      this.messagesReceived++;
      
      // 通知监控Actor消息已接收
      if (this.monitorPid) {
        this.system.root.send(this.monitorPid, { 
          type: 'message-received', 
          messageId,
          latency
        });
      }
      
      return { response, latency };
    } catch (error) {
      console.error(`${this.name} 发送消息失败: ${messageId}`, error);
      this.errors++;
      
      // 通知监控Actor错误发生
      if (this.monitorPid) {
        this.system.root.send(this.monitorPid, { 
          type: 'error', 
          messageId,
          error: error instanceof Error ? error.message : String(error)
        });
      }
      
      return null;
    }
  }
  
  // 获取节点统计信息
  getStats(): any {
    return {
      name: this.name,
      actorCount: this.workerActors.length,
      messagesSent: this.messagesSent,
      messagesReceived: this.messagesReceived,
      errors: this.errors,
      address: this.getAddress()
    };
  }
}

// ====== 测试协调器 ======

class TestCoordinator {
  private nodes: ClusterNode[] = [];
  private allActors: PID[] = [];
  private running: boolean = false;
  private testStartTime: number = 0;
  private testEndTime: number = 0;
  private coordinatorNode: ClusterNode | null = null;
  private monitorPid: PID | null = null;
  
  async setup(nodeCount: number = CONFIG.NODE_COUNT): Promise<void> {
    console.log('======= 设置集群压力测试 =======');
    console.log(`节点数量: ${nodeCount}`);
    console.log(`每节点Actor数量: ${CONFIG.ACTORS_PER_NODE}`);
    console.log(`每Actor消息数量: ${CONFIG.MESSAGES_PER_ACTOR}`);
    console.log(`批次大小: ${CONFIG.BATCH_SIZE}`);
    console.log(`批次间隔: ${CONFIG.BATCH_INTERVAL_MS}ms`);
    console.log(`消息超时: ${CONFIG.MESSAGE_TIMEOUT_MS}ms`);
    console.log('==================================');
    
    // 创建第一个节点作为协调器
    this.coordinatorNode = new ClusterNode('coordinator', 4000, [], true);
    await this.coordinatorNode.start();
    this.nodes.push(this.coordinatorNode);
    
    // 获取监控Actor
    this.monitorPid = this.coordinatorNode.getMonitorPid();
    
    // 获取coordinator的PeerId并构建正确的引导地址
    const coordinatorAddress = this.coordinatorNode.getAddress();
    const peerId = coordinatorAddress.replace('libp2p:', '');
    const bootstrapAddrs = [`/ip4/127.0.0.1/tcp/4000/p2p/${peerId}`];
    
    console.log(`协调器地址: ${coordinatorAddress}`);
    console.log(`引导地址: ${bootstrapAddrs[0]}`);
    
    // 创建其他节点
    for (let i = 1; i < nodeCount; i++) {
      const node = new ClusterNode(`node${i}`, 4000 + i, bootstrapAddrs);
      await node.start();
      this.nodes.push(node);
      console.log(`节点${i}已启动，地址: ${node.getAddress()}`);
    }
    
    // 等待网络连接建立
    console.log('等待网络连接建立...');
    await delay(3000);
    
    // 收集所有Actor
    for (const node of this.nodes) {
      this.allActors.push(...node.getActors());
    }
    
    console.log(`集群中总共有 ${this.allActors.length} 个Actor`);
  }
  
  async runTest(): Promise<any> {
    if (this.allActors.length === 0) {
      throw new Error('没有可用的Actor，请先调用setup()');
    }
    
    console.log('======= 开始集群压力测试 =======');
    this.running = true;
    this.testStartTime = Date.now();
    
    // 设置测试结束时间
    if (CONFIG.TEST_DURATION_MS > 0) {
      setTimeout(() => {
        this.running = false;
        console.log('测试时间到，停止发送新消息');
      }, CONFIG.TEST_DURATION_MS);
    }
    
    // 创建消息发送任务
    const tasks: Promise<void>[] = [];
    
    for (let i = 0; i < this.nodes.length; i++) {
      const sourceNode = this.nodes[i];
      
      tasks.push(
        (async () => {
          let messageCount = 0;
          let successCount = 0;
          let errorCount = 0;
          const totalMessages = CONFIG.MESSAGES_PER_ACTOR * CONFIG.ACTORS_PER_NODE;
          
          while (this.running && messageCount < totalMessages) {
            // 发送一批消息
            const batchPromises: Promise<any>[] = [];
            const batchStartTime = Date.now();
            
            for (let b = 0; b < CONFIG.BATCH_SIZE && messageCount < totalMessages; b++) {
              // 随机选择目标Actor
              const targetActorIndex = Math.floor(Math.random() * this.allActors.length);
              const targetPid = this.allActors[targetActorIndex];
              
              // 随机消息大小 (10-1000字节)
              const dataSize = Math.floor(Math.random() * 990) + 10;
              
              batchPromises.push(
                sourceNode.sendWorkMessage(targetPid, dataSize)
                  .then(result => {
                    if (result && result.response) {
                      successCount++;
                    }
                    return result;
                  })
                  .catch(error => {
                    errorCount++;
                    console.error(`消息发送失败: ${error.message}`);
                    return null;
                  })
              );
              messageCount++;
            }
            
            // 等待当前批次完成
            const results = await Promise.allSettled(batchPromises);
            const batchEndTime = Date.now();
            const batchDuration = batchEndTime - batchStartTime;
            
            // 打印进度和性能指标
            const progress = (messageCount / totalMessages) * 100;
            if (messageCount % (CONFIG.BATCH_SIZE * 5) === 0) {
              const successRate = (successCount / messageCount) * 100;
              console.log(
                `${sourceNode.getAddress()} 进度: ${progress.toFixed(1)}% ` +
                `(${messageCount}/${totalMessages}), ` +
                `成功率: ${successRate.toFixed(1)}%, ` +
                `批次耗时: ${batchDuration}ms`
              );
            }
            
            // 批次间隔
            await delay(CONFIG.BATCH_INTERVAL_MS);
          }
          
          console.log(
            `${sourceNode.getAddress()} 完成消息发送: ` +
            `${messageCount}/${totalMessages}, ` +
            `成功: ${successCount}, ` +
            `失败: ${errorCount}`
          );
        })()
      );
    }
    
    // 等待所有任务完成
    await Promise.all(tasks);
    
    this.testEndTime = Date.now();
    const testDuration = (this.testEndTime - this.testStartTime) / 1000;
    console.log(`测试完成，耗时: ${testDuration.toFixed(2)}秒`);
    
    // 获取最终统计信息
    const stats = await this.getStats();
    console.log('\n========== 测试最终统计 ==========');
    console.log(`总测试时间: ${stats.testDuration.toFixed(2)}秒`);
    console.log(`总消息发送数: ${stats.totalMessagesSent}`);
    console.log(`总消息接收数: ${stats.totalMessagesReceived}`);
    console.log(`总错误数: ${stats.totalErrors}`);
    console.log(`成功率: ${stats.successRate.toFixed(2)}%`);
    console.log(`平均延迟: ${stats.avgLatency ? stats.avgLatency.toFixed(2) : 'N/A'}ms`);
    console.log(`系统吞吐量: ${stats.systemThroughput.toFixed(2)} 消息/秒`);
    console.log('==================================\n');
    
    return stats;
  }
  
  async getStats(): Promise<any> {
    // 从每个节点收集统计信息
    const nodeStats = this.nodes.map(node => node.getStats());
    
    // 从监控Actor获取详细统计信息
    let monitorStats = null;
    if (this.monitorPid && this.coordinatorNode) {
      try {
        monitorStats = await this.coordinatorNode.getSystem().root.requestFuture(
          this.monitorPid,
          { type: 'print-stats' },
          5000
        ).toPromise();
      } catch (error) {
        console.error('获取监控统计信息失败:', error);
      }
    }
    
    // 汇总统计信息
    const totalMessagesSent = nodeStats.reduce((sum, stats) => sum + stats.messagesSent, 0);
    const totalMessagesReceived = nodeStats.reduce((sum, stats) => sum + stats.messagesReceived, 0);
    const totalErrors = nodeStats.reduce((sum, stats) => sum + stats.errors, 0);
    const testDuration = (this.testEndTime - this.testStartTime) / 1000;
    
    // 计算性能指标
    const successRate = totalMessagesSent > 0 
      ? (totalMessagesReceived / totalMessagesSent) * 100 
      : 0;
    const systemThroughput = testDuration > 0 
      ? totalMessagesReceived / testDuration 
      : 0;
    
    // 计算每个节点的性能指标
    const nodePerformance = nodeStats.map(stats => {
      const nodeThroughput = testDuration > 0 
        ? stats.messagesReceived / testDuration 
        : 0;
      const nodeSuccessRate = stats.messagesSent > 0
        ? (stats.messagesReceived / stats.messagesSent) * 100
        : 0;
      
      return {
        name: stats.name,
        address: stats.address,
        actorCount: stats.actorCount,
        messagesSent: stats.messagesSent,
        messagesReceived: stats.messagesReceived,
        errors: stats.errors,
        throughput: nodeThroughput,
        successRate: nodeSuccessRate
      };
    });
    
    // 计算延迟统计
    const latencies = monitorStats?.latencies || [];
    const avgLatency = latencies.length > 0
      ? latencies.reduce((sum: number, latency: number) => sum + latency, 0) / latencies.length
      : null;
    const minLatency = latencies.length > 0
      ? Math.min(...latencies)
      : null;
    const maxLatency = latencies.length > 0
      ? Math.max(...latencies)
      : null;
    
    // 计算每秒消息数的时间序列
    const messageRates = monitorStats?.messageRates || [];
    const avgMessageRate = messageRates.length > 0
      ? messageRates.reduce((sum: number, rate: number) => sum + rate, 0) / messageRates.length
      : 0;
    const peakMessageRate = messageRates.length > 0
      ? Math.max(...messageRates)
      : 0;
      
    // 合并统计信息
    return {
      testDuration,
      nodeCount: this.nodes.length,
      actorCount: this.allActors.length,
      totalMessagesSent,
      totalMessagesReceived,
      totalErrors,
      successRate,
      systemThroughput,
      avgLatency,
      minLatency,
      maxLatency,
      avgMessageRate,
      peakMessageRate,
      nodePerformance,
      ...monitorStats
    };
  }
  
  async shutdown(): Promise<void> {
    console.log('关闭所有节点...');
    for (const node of this.nodes) {
      await node.shutdown();
    }
    console.log('测试协调器已关闭');
  }
}

// ====== 主函数 ======

async function main() {
  try {
    const coordinator = new TestCoordinator();
    
    // 设置集群
    await coordinator.setup();
    
    // 运行测试
    await coordinator.runTest();
    
    // 关闭集群
    await coordinator.shutdown();
    
    console.log('压力测试完成');
  } catch (error) {
    console.error('压力测试失败:', error);
  }
}

// 执行主函数
main().catch(console.error); 